﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using VulnDBExplorer.Framework.Impl;
using System.Threading;
using Luna.Framework;
using System.Diagnostics;

namespace VulnDB.Action.DataCollectors.Actions {
	
	public abstract class MultiThreadDataCollector<T>: DataCollectorAction {

		public int NumberOfWorker { get; set; }

		public int MaxQueueCount { get; set; }

		public Stopwatch Watch { get; protected set; }

		private List<DataCollectorWorker<T>> workers;
		private Queue<T> jobs;
		private Timer progressUpdate;

		protected Queue<T> JobPool { get { return jobs; } }

		public MultiThreadDataCollector() {
			MaxQueueCount = int.MaxValue;
			jobs = new Queue<T>();
		}

		protected void intializeWorkers(int numOfWorker) {
			this.NumberOfWorker = numOfWorker;
			workers = new List<DataCollectorWorker<T>>(numOfWorker);
			for (int i = 0; i < numOfWorker; i++) {
				DataCollectorWorker<T> worker = new DataCollectorWorker<T>(this, jobs);
				workers.Add(worker);
				worker.WorkingThread.Name = Name + ".worker." + i.ToString();
				worker.Execute += new EventHandler<DataCollectorExecutionArg<T>>(jobExecute);
				worker.Completed += new EventHandler<DataCollectorExecutionArg<T>>(jobCompleted);
				worker.Terminated += new EventHandler(jobTerminated);
			}
			Thread.Sleep(100);
		}

		private void jobTerminated(object sender, EventArgs args) {
			lock (workers) {
				workers.Remove(sender as DataCollectorWorker<T>);
				Monitor.PulseAll(workers);
			}
		}

		private void jobCompleted(object sender, DataCollectorExecutionArg<T> e) {
			if (e.Error != null)
				abort(e.Error);
			else {
				onJobCompleted(e.Job, e.Result);
			}
		}

		private void jobExecute(object sender, DataCollectorExecutionArg<T> e) {
			e.Result = onJobExecute(e.Job);
		}

		protected abstract object onJobExecute(T job);

		protected abstract void onJobCompleted(T job, object result);

		protected virtual void onUpdateProgress(object status) {
		}

		protected void startProgressUpdateTimer(int interval) {
			progressUpdate = new Timer(onUpdateProgress, null, 0, interval);
		}

		protected void stopProgressUpdateTimer() {
			if (progressUpdate != null) {
				progressUpdate.Dispose();
				progressUpdate = null;
			}
		}

		protected override void onExecute(object sender, ActionExecutionArgs args) {
			Watch = new Stopwatch();
			setupExecution(args); 

			JobPool.Clear();

			this.intializeWorkers(NumberOfWorker);

			Watch.Start();
			foreach(T job in enumJobs()) {
				if (JobPool.Count >= MaxQueueCount)
					lock (JobPool)
						Monitor.Wait(JobPool);
				lock (JobPool) {
					JobPool.Enqueue(job);
					Monitor.PulseAll(JobPool);
				}
			}

			// add a termination signal
			lock (JobPool) {
				JobPool.Enqueue(default(T));
				Monitor.PulseAll(JobPool);
			}

			// wait for all workers terminated
			while (true)
				lock (workers)
					if (workers.Count > 0)
						Monitor.Wait(workers);
					else break;

			Watch.Stop();
			finalizeExecution();
		}

		protected override void onError(Exception exception) {
			finalizeExecution();
			stopWorkers();
			base.onError(exception);
		}

		protected abstract void setupExecution(ActionExecutionArgs args);

		protected abstract void finalizeExecution();

		protected abstract IEnumerable<T> enumJobs();

		protected void stopWorkers() {
			if (workers != null)
				foreach (DataCollectorWorker<T> worker in workers)
					worker.stop();
		}
	}

	public class DataCollectorWorker<T>  {

		public MultiThreadDataCollector<T> Collectors { get; private set; }
		
		public Thread WorkingThread { get; private set; }

		public Queue<T> JobPool { get; private set; }

		private bool aborted;

		public event EventHandler<DataCollectorExecutionArg<T>> Execute;
		public event EventHandler<DataCollectorExecutionArg<T>> Completed;
		public event EventHandler Terminated;

		public DataCollectorWorker(MultiThreadDataCollector<T> collector, Queue<T> jobPool) {
			this.Collectors = collector;
			this.WorkingThread = new Thread(process);
			this.JobPool = jobPool;

			this.WorkingThread.Start();
		}

		private void process() {
			try {
				while (true) {
					T job = default(T);
					while (!aborted && EqualityComparer<T>.Default.Equals(job, default(T)))
						lock (JobPool) {
							if (JobPool.Count > 0) {
								job = EqualityComparer<T>.Default.Equals(JobPool.Peek(), default(T)) ? default(T) : JobPool.Dequeue();
								aborted = EqualityComparer<T>.Default.Equals(job, default(T));
								Monitor.PulseAll(JobPool);
							}
							else if (!aborted)
								Monitor.Wait(JobPool);
						}
					if (aborted)
						break;
					else {
						DataCollectorExecutionArg<T> e = new DataCollectorExecutionArg<T>(job);
						if (Execute != null)
							Execute(this, e);
						if (Completed != null)
							Completed(this, e);
					}
				}

				if (Terminated != null)
					Terminated(this, EventArgs.Empty);
			}
			catch (Exception ex) {
				if (Completed != null)
					Completed(this, new DataCollectorExecutionArg<T>(default(T)) { Error = ex });
			}
		}

		public void stop() {
			aborted = true;
			WorkingThread.Abort();
		}
	}

	public class DataCollectorExecutionArg<T> : EventArgs {
		public T Job { get; private set; }
		public object Result { get; set; }
		public Exception Error { get; set; }

		public DataCollectorExecutionArg(T job) {
			this.Job = job;
			Error = null;
			Result = null;
		}
	}
}
